package chapter1;

import Config.GlobalKafkaConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;


public class ProducerFastStart {
    public static final String brokerList = GlobalKafkaConfig.brokerList;
    public static final String topic = GlobalKafkaConfig.topic;


    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("bootstrap.servers", brokerList);


        KafkaProducer<String, String> producer =
                new KafkaProducer<>(properties);

        try {

            for (int i = 0; i < 10; i++) {
                ProducerRecord<String, String> record =
                        new ProducerRecord<>(topic, "hello, Kafka!"+i);
                RecordMetadata recordMetadata = producer.send(record).get();
                System.out.println("hasOffset:  "+recordMetadata.hasOffset());
                System.out.println("hasTimestamp:  "+recordMetadata.hasTimestamp());
                System.out.println("offset:  "+recordMetadata.offset());
                System.out.println("partition:  "+recordMetadata.partition());
                System.out.println("timestamp:  "+recordMetadata.timestamp());
                System.out.println("topic:  "+recordMetadata.topic());
                System.out.println(recordMetadata.toString());
                System.out.println("======================================");
                /**
                 * ======================================
                 * hasOffset:  true
                 * hasTimestamp:  true
                 * offset:  12
                 * partition:  1
                 * timestamp:  1604900932673
                 * topic:  wang
                 * wang-1@12
                 */
            }

//            producer.send(record).get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        producer.close();
    }
}
